package com.lac.acme.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.jupiter.api.Test;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

public class SimpleProducer {


    @Test
    public void producerTest() {
        Map<String, Object> producerPrep = new HashMap<>();
        producerPrep.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.60.134:9092");
        producerPrep.put(ProducerConfig.ACKS_CONFIG, "1");
        producerPrep.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        producerPrep.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        producerPrep.put(ProducerConfig.BATCH_SIZE_CONFIG, 1024 * 1024);
        producerPrep.put(ProducerConfig.LINGER_MS_CONFIG, 1000 * 10);

        byte[] inputBuf = new byte[1024];
        try (KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(producerPrep);) {
            while (true) {
                int readLen = System.in.read(inputBuf);
                String inputStr = new String(inputBuf, 0, readLen, StandardCharsets.UTF_8);
                inputStr = inputStr.replaceAll("\\n", "");

                String messageValue = String.format("%s:%s", System.currentTimeMillis(), inputStr);
                ProducerRecord<String, String> producerRecord = new ProducerRecord<>("TOPIC_A",  "key_a:" +readLen , messageValue);

                kafkaProducer.send(producerRecord, (metadata, exception) -> {
                    long offset = metadata.offset();
                    int partition = metadata.partition();
                    String topic = metadata.topic();
                    System.out.printf("topic:%s partition:%s offset:%s %s", topic, partition, offset, "\n");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
